package com.gzsxy.esjy.message.dispatch;


import com.gzsxy.esjy.bus.message.InnerProtocol;
import com.gzsxy.esjy.bus.message.MessageFactory;
import com.gzsxy.esjy.message.channel.ChannelManager;
import com.gzsxy.esjy.message.handler.MessageCenterService;
import com.gzsxy.esjy.message.netty.Message;
import io.netty.channel.Channel;
import io.netty.channel.group.ChannelGroup;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.util.HashMap;
import java.util.Map;
import java.util.function.Function;

/**
 * 默认分派器
 * @author zhangxiaolong
 *
 */
@Component
@Slf4j
public class DefaultMessageDispatch implements MessageDispatch{

	@Value("${spring.application.name}")
	private String application ;

	/**
	 * 全局主题，所有连接到message-centen的websocket的通道都会订阅这个主题
	 */
	@Value("${spring.cloud.stream.bindings.default-in-broadcast.destination:broadcast}")
	private String broadcast ;

	@Autowired
	private MessageFactory<InnerProtocol> factory ;
	
	@Autowired
	private ChannelManager channelManager ;
	
	@Autowired
	private MessageCenterService messageCenterService ;

	/**
	 * 消息转换器
	 */
	private Map<String, Function<InnerProtocol,Object>> messageConver = new HashMap<>() ;

	/**
	 * 目前仅支持只定义了/mds协议的通信
	 */
	@Override
	public void busToWebSocket(InnerProtocol message,String dest) {
		String cmd = message.getCmd();
	
		ChannelGroup group = null;
		if(dest.equals(broadcast)) {
			group  = channelManager.getChannel(broadcast) ;
		}else {
			group = channelManager.getChannel(cmd) ;
		}
		if(group == null) {
			log.warn("无客户端订阅此主题【{}】消息", cmd);
			return ;
		}
		Function<InnerProtocol, Object> convert = messageConver.get(cmd);
		if(convert != null){
			group.writeAndFlush(convert.apply(message));
		}else{
			//去除个性化处理去除后缀
			message.setCmd(stripUri(cmd,message.getStrip()));
			group.writeAndFlush(Message.convert(message)) ;
		}
	}

	private String stripUri(String cmd, int strip){
		String[] split = cmd.split("/");
		if(split.length > 1 && strip > 0 &&split.length > strip){
			StringBuilder builder = new StringBuilder() ;
			int length = split.length-strip;
			for (int i = 0; i < length; i++) {
				builder.append(split[i]) ;
				if(i != length-1){
					builder.append("/") ;
				}
			}
			return builder.toString() ;
		}
		return cmd ;
	}
	
	@Override
	public void webSocketToBus(Channel channel, Message message) {
		String cmd = message.getCmd();
		//协议cmd格式 订阅cmd=message-center/subscrib  取消订阅cmd=message-center/unsubscrib
		if(cmd.startsWith(application.toLowerCase())) {
			//最终message.getBody()作为主题，进行订阅  这里可以存放具体的服务名，
			// 当bus传数据过来需要根据服务名传输到websocket时可以根据这个设计
			messageCenterService.service(message, channel);
			return ;
		}
		//如果不是以上这种，就属于bus发送指定服务进行通信
		int index = cmd.indexOf("/") ;
		if(index > 0) {
			cmd = cmd.substring(0, index) ;
		}else {
			channel.writeAndFlush(Message.warn("cmd 命令格式错误 ")) ;
			return ;
		}
		InnerProtocol protocol = new InnerProtocol() ;
		protocol.setBody(message.getBody());
		protocol.setCmd(message.getCmd().replace(cmd, ""));
		factory.send(cmd, protocol);
		channel.writeAndFlush(Message.success(null)) ;
	}


	@Override
	public void channelCloseEvent(Channel channel) {
		//添加通道
		channelManager.addChannel(broadcast, channel);
		//绑定关闭事件
		channelManager.closeEvent(channel);
	}

	@Override
	public void addSubscrib(String topic,Channel channel,boolean removeBroadcast){
		if(removeBroadcast){
			//将当前通道从broadcast主题中移除
			channelManager.getChannel(broadcast).remove(channel);
		}
		//重新订阅新的主题，通道
		channelManager.addChannel(topic, channel);
	}

	@Override
	public void addMessageConver(String topic, Function<InnerProtocol, Object> function) {
		messageConver.put(topic,function) ;
	}

}
